package org.example.hadoop3;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;

/**
 * job.setOutputFormatClass()
 */

public class LogRecordWriter extends RecordWriter<Text, NullWritable> {
    private String gPath = "";
    private String qPath = "";
    private FileSystem fs;
    private FSDataOutputStream gStream;
    private FSDataOutputStream qStream;

    public LogRecordWriter(TaskAttemptContext context) throws IOException {
        fs = FileSystem.get(context.getConfiguration());
        gStream = fs.create(new Path(gPath));
        qStream = fs.create(new Path(qPath));
    }

    @Override
    public void write(Text text, NullWritable nullWritable) throws IOException, InterruptedException {
        String str = text.toString();
        if (str.contains("gaoqiang")) {
            gStream.writeBytes(str + "\r");
        } else if (str.contains("pang")) {
            qStream.writeBytes(str + "\r");
        }
    }

    @Override
    public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        IOUtils.closeStream(gStream);
        IOUtils.closeStream(qStream);
    }
}
